package com.codejiwei.flink.table;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.LocalDateTime;

/**
 * @author jiwei
 * @description
 * @date 2023/8/15 17:21
 */
public class Flink_Table_API_Table2Stream {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema = Schema.newBuilder()
                .column("id", DataTypes.INT().notNull())
                .column("name", DataTypes.STRING())
                .column("ts", DataTypes.TIMESTAMP(3))
                .primaryKey("id")
                .watermark("ts", "ts - INTERVAL '3' SECOND")
                .build();

        tableEnv.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(schema)
//                .option("rows-per-second", "1")
                .option("number-of-rows", "10")
                .option("fields.name.length", "5")
                .option("fields.id.min", "0")
                .option("fields.id.max", "10")
                .build());

        Table table = tableEnv.from("SourceTable");
        SingleOutputStreamOperator<User> map = tableEnv.toDataStream(table)
                .map(new MapFunction<Row, User>() {
                    @Override
                    public User map(Row value) throws Exception {
                        Integer id = value.<Integer>getFieldAs("id");
                        String name = value.<String>getFieldAs("name");
                        LocalDateTime ts = value.<LocalDateTime>getFieldAs("ts");
                        return new User(id, name, ts, "aaa");
                    }
                });

//        map.print();

        Schema schema1 = Schema.newBuilder()
                .column("f0", DataTypes.of(User.class))
                .build();

        Table table1 = tableEnv.fromDataStream(map, schema1);

        tableEnv.createTemporaryView("tempView", table1);

        //Sink
        tableEnv.executeSql("create table sinkTable (id bigint, name string, ts timestamp(3), AAA string) " +
                "with ('connector' = 'print')");

        tableEnv.executeSql("insert into sinkTable select * from tempView");

        env.execute();
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class User{
    public Integer id;

    public String name;

    public LocalDateTime ts;

    public String AAA;

}